package fun.xinnix.utils;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Slf4j
@Component
public class KafkaUtil {
    @Resource
    KafkaTemplate<Object, Object> kafkaTemplate;

    public void sendMessage(String topicName, Object obj) {
        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(topicName, obj);
        sendCallback(future);
    }

    public void sendMessage(String topicName, String key, Object obj) {
        if (StringUtils.isBlank(key)) {
            sendMessage(topicName, obj);
            return;
        }
        ListenableFuture<SendResult<Object, Object>> future = kafkaTemplate.send(topicName, key, obj);
        sendCallback(future);
    }

    private void sendCallback(ListenableFuture<SendResult<Object, Object>> future) {
        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }

            @Override
            public void onSuccess(SendResult<Object, Object> result) {
                log.info("send successful:{}", result);
            }
        });
    }
}
